package com.zhuanzhuan.youpin.handle;

import java.lang.reflect.Method;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.AfterReturning;
import org.aspectj.lang.annotation.AfterThrowing;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.rocketmq.common.message.MessageExt;
import com.bj58.pigeon.common.util.IDHelper;
import com.bj58.youpin.common.utils.json.JsonUtils;
import com.zhuanzhuan.youpin.annotation.ConsumerExceptionAction;
import com.zhuanzhuan.youpin.entity.EHandleStatus;
import com.zhuanzhuan.youpin.entity.FailedMsg;
import com.zhuanzhuan.youpin.exception.ConsumeException;
import com.zhuanzhuan.youpin.service.IConsumerService;
import com.zhuanzhuan.youpin.utils.MD5;

/**
 * @author zz.youpin.G_Y
 * @date 2018年6月14日
 * @说明:mq消费consume方法切面处理
 * 入参
 */
@Aspect
@Component
public class ConsumerExceptionAspect {

    private static ThreadPoolExecutor threadPool =
            new ThreadPoolExecutor(3, 6, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(1000), new ThreadPoolExecutor.CallerRunsPolicy());

    @Autowired
    private IConsumerService consumerService;

    private static Logger logger = LoggerFactory.getLogger(ConsumerExceptionAspect.class);

    @Pointcut("@annotation(com.zhuanzhuan.youpin.annotation.ConsumerExceptionAction)")
    public void annotationPointCut() {};

    /**
     * @param point
     * @param e
     * @throws Exception
     * @author zz.youpin.G_Y
     * @date 2018年6月14日
     * @说明:mq消费异常
     */
    @AfterThrowing(pointcut = "annotationPointCut()", throwing = "e")
    public void handle(final JoinPoint point, Exception e) {

        logger.info("-------------ConsumerExceptionAspect AfterThrowing bigin-------------");

        ConsumeException consumeException = null;
        if (e instanceof ConsumeException) {
            e = (ConsumeException) e;
            consumeException = (ConsumeException) e;
        }
        final ConsumeException consumeExceptionF = consumeException;

        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                MethodSignature signature = (MethodSignature) point.getSignature();
                Method method = signature.getMethod();
                ConsumerExceptionAction action = method.getAnnotation(ConsumerExceptionAction.class);
                logger.info("AfterThrowing 注解式拦截 :{} ", action.name());
                FailedMsg failedMsg = null;
                if (consumeExceptionF == null) {
                    // 异常非ConsumeException异常 需要从 JoinPoint 获取args中mq信息
                    failedMsg = getFailedMsgFrom(point);
                    if (failedMsg == null) {
                        return;
                    }
                } else {
                    // 异常为ConsumeException异常 可以直接从异常中获取mq信息
                    failedMsg = new FailedMsg().setId(IDHelper.genshortId(0)).setBody(consumeExceptionF.getBody())
                            .setBodyHash(MD5.encode(consumeExceptionF.getBody())).setCreateTime(System.currentTimeMillis())
                            .setMsgkeys(consumeExceptionF.getKey()).setStatus(EHandleStatus.FAILED_STATUS.status()).setTimes(0)
                            .setTopicAndTags(consumeExceptionF.getMessageExt().getTopic() + "|" + consumeExceptionF.getMessageExt().getTags());
                }
                logger.info("failedMsg.getStatus={} ", failedMsg.getStatus());
                logger.info("-------begin consumerService.addFailedMsg---------==={} ", failedMsg.toString());
                consumerService.addFailedMsg(failedMsg);
                logger.info("-------------Add FailedMsg success-------------");
            }
        });
        // junit test need open
        /*try {
            TimeUnit.SECONDS.sleep(1L);
        } catch (InterruptedException e1) {
            // TODO Auto-generated catch block
            e1.printStackTrace();
        }*/
    }

    /**
     * @param joinPoint
     * @param rvt
     * @author zz.youpin.G_Y
     * @date 2018年6月13日
     * @说明:mq正常return
     * 可能为true处理成功
     * 可能为发false处理失败
     */
    @AfterReturning(pointcut = "annotationPointCut()", returning = "rvt")
    public void afterReturn(final JoinPoint joinPoint, final boolean rvt) {

        logger.info("----------------afterReturn begin---------------");

        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                MethodSignature signature = (MethodSignature) joinPoint.getSignature();
                Method method = signature.getMethod();
                ConsumerExceptionAction action = method.getAnnotation(ConsumerExceptionAction.class);
                logger.info("AfterReturning 注解式拦截 :{} ", action.name());
                FailedMsg failedMsg = getFailedMsgFrom(joinPoint);
                if (failedMsg == null) {
                    return;
                }
                if (rvt) {
                    logger.info("afterReturn SUCCESS");
                    // 需要将存在的FailedMsg置成功态
                    consumerService.onecMqSuccess(failedMsg.getTopicAndTags(), failedMsg.getBodyHash());
                } else {
                    logger.info("afterReturn FAILED");
                    // 新增不存在的FailedMsg
                    consumerService.addFailedMsg(failedMsg);
                }
                logger.info("----------------afterReturn end---------------");
            }
        });
        // junit test need open
        /*try {
            TimeUnit.SECONDS.sleep(1L);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }*/
    }

    /**
     * 
     * @param point
     * @return
     * @author zz.youpin.G_Y
     * @date 2018年6月13日
     * @说明:从JoinPoint中获取msg内容
     * args中 第一个参数为MessageExt
     * 第二个参数为body
     * 第三个参数为keys
     */
    private FailedMsg getFailedMsgFrom(JoinPoint point) {
        Object[] args = point.getArgs();
        if (args == null || args.length < 2) {
            logger.error("Add FailedMsg error ! The reason is point.getArgs().length={} ", args == null ? null : args.length);
        }
        try {
            FailedMsg failedMsg = new FailedMsg();
            MessageExt messageExt = (MessageExt) args[0];
            String bodyJson = null;
            if (args[1] instanceof String) {
                bodyJson = (String) args[1];
                if (StringUtils.isEmpty(bodyJson)) {
                    return null;
                }
            } else {
                bodyJson = JsonUtils.toJson(args[1]).get();
            }
            String keys = (String) args[2];
            return failedMsg.setId(IDHelper.genshortId(0)).setBody(bodyJson).setBodyHash(MD5.encode(bodyJson))
                    .setCreateTime(System.currentTimeMillis()).setMsgkeys(keys).setStatus(EHandleStatus.FAILED_STATUS.status()).setTimes(0)
                    .setTopicAndTags(messageExt.getTopic() + "|" + messageExt.getTags());
        } catch (Exception e) {
            logger.error("Add FailedMsg error ! The reason is class change Exception ! args.length={}  e={} ", args == null ? null : args.length,
                    e.getMessage(), e);
        }
        return null;
    }

}
